package org.example.testcode.helper;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.RequestLine;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.nested.ParsedNested;
import org.elasticsearch.search.aggregations.bucket.terms.*;
import org.elasticsearch.search.aggregations.metrics.InternalAvg;
import org.elasticsearch.search.aggregations.metrics.ParsedAvg;
import org.elasticsearch.search.aggregations.metrics.Sum;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.example.testcode.annotation.ESDocObject;
import org.example.testcode.annotation.ESField;
import org.example.testcode.annotation.EnableSuggest;
import org.example.testcode.bean.EsGoodsInfo;
import org.example.testcode.bean.EsProperties;
import org.example.testcode.bean.TopicEsInfo;
import org.example.testcode.enumeration.ESAnalyzer;
import org.example.testcode.enumeration.ESFieldType;
import org.hibernate.validator.internal.util.StringHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.util.*;

import static org.elasticsearch.index.query.QueryBuilders.queryStringQuery;


/**
 * es通用模板类
 */

@Slf4j
@Component
@Configuration
public class EsTemplate {

//    @Autowired
//    private RestHighLevelClient client;

    public static final String topicIndex = "techtopic";

    private static final ObjectMapper MAPPER = new ObjectMapper();

    private RestHighLevelClient client; //高级客户端
    private RestHighLevelClient remoteClient;
//================================ 初始化连接 ==============================

    //public static String remoteEsHost = "39.108.226.6";
    public static String remoteEsHost = "192.168.137.200";
    public static String esHost = "192.168.137.201";
    //public static String esHost = "localhost";


    @Autowired
    private EsProperties esProperties;

    /**
     * 初始化连接
     */
    @PostConstruct
    public void initEs() {
        try {
            //String host = esProperties.getHost();
            //初始化：高级客户端
            client = new RestHighLevelClient(RestClient.builder(new HttpHost(esHost, 9200, "http"),
                    new HttpHost(esHost, 9201, "http")));
            System.out.println("[" + esHost + "]成功启动Elasticsearch服务");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("[" + esHost + "]启动Elasticsearch服务失败");
        }

        //远程客户端
        try {
            remoteClient = new RestHighLevelClient(RestClient.builder(new HttpHost(remoteEsHost, 9200, "http"),
                    new HttpHost(remoteEsHost, 9201, "http")));
            System.out.println("[" + remoteEsHost + "]成功启动Elasticsearch服务");
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("[" + remoteEsHost + "]启动Elasticsearch服务失败");
        }
    }

    /**
     * 关闭连接
     */
    public void closeEs() {
        try {
            client.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            remoteClient.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 构建对Elasticsearch服务的请求
     */
    public void executeRequest() {
        Request request = new Request("GET", "/"); //在服务器上请求

        try {
            //方法一：使用performRequest同步请求方法
            //Response response = restClient.performRequest(request);
            Response response = client.getLowLevelClient().performRequest(request);

            //获取响应结果内容：
            //已执行请求的信息
            RequestLine requestLine = response.getRequestLine();
            System.out.println("请求的信息：" + requestLine.getUri());

            //Host返回的信息
            HttpHost host = response.getHost();
            System.out.println("Host返回的信息：" + host.toString());

            //响应状态，从中可以解析状态代码
            int statusCode = response.getStatusLine().getStatusCode();
            System.out.println("响应状态：" + statusCode);

            //响应头，也可以通过getheader(string)按名称获取
            Header[] headers = response.getHeaders();
            String responseBody = EntityUtils.toString(response.getEntity());
            System.out.println("响应头：" + responseBody);
            System.out.println("响应结果：" + response.toString());

            //方法二：使用performRequestAsync异步请求方法
            //使用异步时，下面的关闭方法要注释掉
            /*restClient.performRequestAsync(request, new ResponseListener()
            {
                @Override
                public void onSuccess(Response response)
                {
                    System.out.println("使用performRequestAsync异步请求方法得到结果：" + response.toString());
                }
                @Override
                public void onFailure(Exception e)
                {
                    System.out.println("错误");
                }
            });*/
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeEs();
        }
    }


    private static final String TIMESTAMP = "timestamp";


    /**
     * =========================================    mapping    ===============================================
     **/

    private static XContentBuilder createBuilder() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        builder.startObject("properties");
        builder.startObject("message");
        builder.field("type", "text");
        // 为message字段设置分词器
//                    builder.field("analyzer", "ik_smart");
        builder.endObject();

        builder.startObject("timestamp");
        builder.field("type", "dateTime");
        // 设置日期时间格式为Long类型
        builder.field("format", "epoch_millis");
        builder.endObject();
        builder.endObject();
        builder.endObject();
        return builder;
    }


    /**
     * 创建索引(默认分片数为5和副本数为1)
     *
     * @param indexName
     * @param esMapTypeClass
     * @throws IOException
     */
    public boolean createIndex(String indexName, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        if (checkIndexExists(indexName)) {
            log.error("\"index={}\"索引已经存在！", indexName);
            return false;
        }
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        XContentBuilder builder = buildXContentBuilder(indexName, esMapTypeClass);
        request.mapping(esMapTypeClass.getSimpleName(), builder);

        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        // 指示是否所有节点都已确认请求
        boolean acknowledged = response.isAcknowledged();
        // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            log.info("创建索引成功！索引名称为{}", indexName);
            return true;
        }

        log.error("创建索引失败");
        return false;

    }

    /**
     * 创建索引
     *
     * @param index
     * @throws IOException
     */
    public boolean createIndex(String index) throws IOException {
        if (checkIndexExists(index)) {
            log.error("\"index={}\"索引已存在！", index);
            return false;
        }
        CreateIndexRequest request = new CreateIndexRequest(index);
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        boolean acknowledged = response.isAcknowledged();
        boolean shardsAcknowledged = response.isShardsAcknowledged();
        if (acknowledged || shardsAcknowledged) {
            log.info("创建索引成功！索引名称为{}", index);
            return true;
        } else {
            log.error("创建索引失败");
            return false;
        }
    }

    /**
     * 创建索引(传入参数：分片数、副本数)
     *
     * @param indexName
     * @param shards
     * @param replicas
     * @throws IOException
     */
    public boolean createIndex(String indexName, int shards, int replicas) throws IOException {
        if (checkIndexExists(indexName)) {
            log.error("\"index={}\"索引已存在！", indexName);
            return false;
        }
        Settings.Builder builder = Settings.builder().put("index.number_of_shards", shards).put("index.number_of_replicas", replicas);
        CreateIndexRequest request = new CreateIndexRequest(indexName).settings(builder);
//        request.mapping(TYPE, generateBuilder());
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            log.info("创建索引成功！索引名称为{}", indexName);
            return true;
        } else {
            log.error("创建索引失败");
            return false;
        }
    }

    /**
     * 删除索引
     *
     * @param indexName
     * @throws IOException
     */
    public Boolean deleteIndex(String indexName) throws IOException {
        try {
            AcknowledgedResponse response = client.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
            if (response.isAcknowledged()) {
                log.info("{} 索引删除成功！", indexName);
                return true;
            }
        } catch (ElasticsearchException ex) {
            if (ex.status() == RestStatus.NOT_FOUND) {
                log.error("{} 索引名不存在", indexName);
                return false;
            }
            log.error("删除失败！");
        }
        return false;
    }


    /**
     * 判断索引是否存在
     *
     * @param indexName
     * @return
     * @throws IOException
     */
    public boolean checkIndexExists(String indexName) {
        GetIndexRequest request = new GetIndexRequest().indices(indexName);
        try {
            return client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("判断索引是否存在，操作异常！");
        }
        return false;
    }


    /**
     * 开启索引
     *
     * @param indexName
     * @throws IOException
     */
    public void openIndex(String indexName) throws IOException {
        if (!checkIndexExists(indexName)) {
            log.error("索引不存在！");
            return;
        }
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        OpenIndexResponse response = client.indices().open(request, RequestOptions.DEFAULT);
        if (response.isAcknowledged() || response.isShardsAcknowledged()) {
            log.info("{} 索引开启成功！", indexName);
        }
    }


    /**
     * 查找文档
     *
     * @param index
     * @param type
     * @param id
     * @return
     * @throws IOException
     */
    public Map<String, Object> getDocument(String index, String type, String id) throws IOException {
        Map<String, Object> resultMap = new HashMap<>();
        GetRequest request = new GetRequest(index, type, id);
        // 实时(否)
        request.realtime(false);
        // 检索之前执行刷新(是)
        request.refresh(true);

        GetResponse response = null;
        try {
            response = client.get(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("文档未找到，请检查参数！");
            }
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("查找失败！");
        }

        if (Objects.nonNull(response)) {
            if (response.isExists()) { // 文档存在
                resultMap = response.getSourceAsMap();
            } else {
                // 处理未找到文档的方案。 请注意，虽然返回的响应具有404状态代码，但仍返回有效的GetResponse而不是抛出异常。
                // 此时此类响应不持有任何源文档，并且其isExists方法返回false。
                log.error("文档不存在！");
            }
        }
        return resultMap;
    }

    /**
     * 判断文档是否存在
     *
     * @param
     * @return true:存在  false:不存在
     * @author hucm
     * @since 2021/5/31
     */
    public boolean existDoc(String index, String type, String id) throws IOException {
        GetRequest request = new GetRequest(index, type, id);
        // 不需要获取source内容
        request.fetchSourceContext(new FetchSourceContext(false));
        return client.exists(request, RequestOptions.DEFAULT);
    }

    /**
     * 删除文档
     *
     * @param indexName
     * @param typeName
     * @param docId
     * @throws IOException
     */
    public void deleteDocument(String indexName, String typeName, String docId) throws IOException {
        DeleteRequest request = new DeleteRequest(indexName, typeName, docId);
        DeleteResponse response = null;
        try {
            response = client.delete(request, RequestOptions.DEFAULT);
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("删除失败!");
        }
        if (Objects.nonNull(response)) {
            if (response.getResult() == DocWriteResponse.Result.NOT_FOUND) {
                log.error("不存在该文档，请检查参数！");
            }
            log.info("文档已删除！");
            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("失败原因：{}", reason);
                }
            }
        }
    }

    /**
     * 通过一个脚本语句(如："ctx._source.posttime=\"2018-09-18\"")更新文档
     * 针对单个字段更新或object类型字段更新
     *
     * @param index
     * @param type
     * @param id
     * @param script
     */
    public void updateDocByScript(String index, String type, String id, String script) throws IOException {
        Script inline = new Script(script);
        UpdateRequest request = new UpdateRequest(index, type, id).script(inline);
        try {
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                log.info("文档更新成功！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                log.error("\"index={},type={},id={}\"的文档已被删除，无法更新！", response.getIndex(), response.getType(), response.getId());
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                log.error("操作没有被执行！");
            }

            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("未处理原因：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("不存在这个文档，请检查参数！");
            } else if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突异常！");
            }
            log.error("更新失败！：{}，脚本：{}，id：{}", e.getMessage(), script, id);
        }
    }

    /**
     * 通过一个脚本语句(针对于object和nested数据类型字段更新)更新文档
     *
     * @param index
     * @param type
     * @param id
     * @param script
     * @param params
     * @throws IOException
     */
    public void updateDocByScript(String index, String type, String id, String script, Map<String, Object> params) throws IOException {
        Script inline = new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG, script, params);
        UpdateRequest request = new UpdateRequest(index, type, id).script(inline);
        try {
            UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
            if (response.getResult() == DocWriteResponse.Result.UPDATED) {
                log.info("文档更新成功！");
            } else if (response.getResult() == DocWriteResponse.Result.DELETED) {
                log.error("\"index={},type={},id={}\"的文档已被删除，无法更新！", response.getIndex(), response.getType(), response.getId());
            } else if (response.getResult() == DocWriteResponse.Result.NOOP) {
                log.error("操作没有被执行！");
            }

            ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                log.error("部分分片副本未处理");
            }
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                    log.error("未处理原因：{}", reason);
                }
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("不存在这个文档，请检查参数！");
            } else if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突异常！");
            }
            log.error("更新失败！:{}，脚本：{}，id：{}", e.getMessage(), script, id);
        }
    }

    /**
     * 批量增加文档(6版本，6+版本)
     *
     * @param indexName
     * @param typeName
     * @param esMappingTypes 添加的文档列表
     * @throws IOException
     */
    public List<String> bulkAdd(String indexName, String typeName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (!CollectionUtils.isEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                IndexRequest request = new IndexRequest(indexName, typeName, mappingType.generateDocId()).source(JSON.toJSONString(mappingType), XContentType.JSON);
                bulkRequest.add(request);
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        // 刷新策略
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量增加操作失败！");
            return new ArrayList<>();
        }
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        // 全部操作成功
        if (!bulkResponse.hasFailures()) {
//            log.info("批量增加操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档增加失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("增加失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("新增失败的货品id：{}", failureList);
                    return failureList;
                } else {
//                    log.info("\"index={}, type={}, id={}\"的文档增加成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }
            }
        }
        return new ArrayList<>();
    }

    /**
     * 批量增加文档(7.6版本，7+版本)
     *
     * @param indexName
     * @param esMappingTypes 添加的文档列表
     * @throws IOException
     */

    public List<String> bulkAdd(String indexName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (!CollectionUtils.isEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                IndexRequest request = new IndexRequest(indexName).id(mappingType.generateDocId()).source(MAPPER.writeValueAsString(mappingType), XContentType.JSON);
                bulkRequest.add(request);
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        // 刷新策略
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量增加操作失败！");
            return new ArrayList<>();
        }
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        // 全部操作成功
        if (!bulkResponse.hasFailures()) {
//            log.info("批量增加操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, id={}\"的文档增加失败！", failure.getIndex(), failure.getId());
                    log.error("增加失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("新增失败的货品id：{}", failureList);
                    return failureList;
                } else {
//                    log.info("\"index={}, type={}, id={}\"的文档增加成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }
            }
        }
        return new ArrayList<>();
    }

    public List<String> bulkAddByRemote(String indexName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (!CollectionUtils.isEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                IndexRequest request = new IndexRequest(indexName).id(mappingType.generateDocId()).source(MAPPER.writeValueAsString(mappingType), XContentType.JSON);
                bulkRequest.add(request);
            }
        }
        // 超时时间(2分钟)
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        // 刷新策略
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);

        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量增加操作失败！");
            return new ArrayList<>();
        }
        BulkResponse bulkResponse = remoteClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        // 全部操作成功
        if (!bulkResponse.hasFailures()) {
//            log.info("批量增加操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, id={}\"的文档增加失败！", failure.getIndex(), failure.getId());
                    log.error("增加失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("新增失败的货品id：{}", failureList);
                    return failureList;
                } else {
//                    log.info("\"index={}, type={}, id={}\"的文档增加成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }
            }
        }
        return new ArrayList<>();
    }


    /**
     * 根据条件删除es数据
     *
     * @param indexName           索引名称
     * @param bigBoolQueryBuilder
     * @return 删除条数
     */
    public Long delByEsQuery(String indexName, BoolQueryBuilder bigBoolQueryBuilder) {
        DeleteByQueryRequest request =
                new DeleteByQueryRequest(indexName);
        request.setQuery(bigBoolQueryBuilder);
        try {
            BulkByScrollResponse bulkResponse =
                    client.deleteByQuery(request, RequestOptions.DEFAULT);
            return bulkResponse.getDeleted();
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.CONFLICT) {
                log.error("版本冲突！");
            }
            log.error("删除失败!:{}", e.getMessage());
        } catch (IOException ioExp) {
            log.error("删除失败:{}", ioExp.getMessage());
        }
        return 0L;

    }

    /**
     * 批量更新文档
     *
     * @param indexName
     * @param typeName
     * @param esMappingTypes
     * @throws IOException
     */
    public List<String> bulkUpdate(String indexName, String typeName, List<? extends IESMappingType> esMappingTypes) throws IOException {
        List<String> failureList = new ArrayList<>();
        BulkRequest bulkRequest = new BulkRequest();
        if (!CollectionUtils.isEmpty(esMappingTypes)) {
            for (IESMappingType mappingType : esMappingTypes) {
                UpdateRequest request = new UpdateRequest(indexName, typeName, mappingType.generateDocId())
                        .doc(JSON.toJSONString(mappingType), XContentType.JSON);
                // true，表明如果文档不存在，则新更新的文档内容作为新的内容插入文档，
                //这个和scriptedUpsert的区别是：更新文档的两种不同方式，有的使用doc方法更新有的使用脚本更新
                request.docAsUpsert(true);

                // 是否将文档内容作为结果返回，默认是禁止的
                request.fetchSource(false);
                bulkRequest.add(request);
            }
        }


        if (bulkRequest.numberOfActions() == 0) {
            log.error("参数错误，批量更新操作失败！");
            return new ArrayList<>();
        }
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (!bulkResponse.hasFailures()) {
//            log.info("批量更新操作成功！");
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档更新失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("更新失败详情: {}", failure.getMessage());
                    failureList.add(failure.getId());
                    log.error("更新失败的货品id: {}", failureList);
                    return failureList;
                }/* else {
                    log.info("\"index={}, type={}, id={}\"的文档更新成功！", bulkItemResponse.getIndex(), bulkItemResponse.getType(), bulkItemResponse.getId());
                }*/
            }
        }
        return failureList;
    }

    /**
     * 批量删除文档
     *
     * @param indexName
     * @param typeName
     * @param docIds
     * @throws IOException
     */
    public boolean bulkDelete(String indexName, String typeName, List<String> docIds) throws IOException {

        BulkRequest bulkRequest = new BulkRequest();
        if (CollectionUtils.isEmpty(docIds)) {
            for (String docId : docIds) {
                DeleteRequest request = new DeleteRequest(indexName, typeName, docId);
                bulkRequest.add(request);
            }
        }
        if (bulkRequest.numberOfActions() == 0) {
            log.error("操作失败，请检查参数！");
            return false;
        }
        bulkRequest.timeout(TimeValue.timeValueMinutes(2L));
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (!bulkResponse.hasFailures()) {
            log.info("批量删除操作成功！");
            return true;
        } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
                if (bulkItemResponse.isFailed()) {
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                    log.error("\"index={}, type={}, id={}\"的文档删除失败！", failure.getIndex(), failure.getType(), failure.getId());
                    log.error("删除失败详情: {}", failure.getMessage());
                    return false;
                }
            }
            return false;
        }
    }

    /**
     * 批量查找文档
     *
     * @param indexName
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> multiGet(String indexName, String typeName, List<String> docIds) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        MultiGetRequest request = new MultiGetRequest();
        if (!CollectionUtils.isEmpty(docIds)) {
            for (String docId : docIds) {
                request.add(new MultiGetRequest.Item(indexName, typeName, docId));
            }
        }
        request.realtime(false);
        request.refresh(true);
        MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
        List<Map<String, Object>> list = parseMGetResponse(response);
        if (!list.isEmpty()) {
            resultList.addAll(list);
        }
        return resultList;
    }

    private List<Map<String, Object>> parseMGetResponse(MultiGetResponse response) {
        List<Map<String, Object>> list = new ArrayList<>();
        MultiGetItemResponse[] responses = response.getResponses();
        for (MultiGetItemResponse item : responses) {
            GetResponse getResponse = item.getResponse();
            if (Objects.nonNull(getResponse)) {
                if (!getResponse.isExists()) {
                    log.error("\"index={}, type={}, id={}\"的文档查找失败，请检查参数！", getResponse.getIndex(), getResponse.getType(), getResponse.getId());
                } else {
                    list.add(getResponse.getSourceAsMap());
                }
            } else {
                MultiGetResponse.Failure failure = item.getFailure();
                ElasticsearchException e = (ElasticsearchException) failure.getFailure();
                if (e.status() == RestStatus.NOT_FOUND) {
                    log.error("\"index={}, type={}, id={}\"的文档不存在！", failure.getIndex(), failure.getType(), failure.getId());
                } else if (e.status() == RestStatus.CONFLICT) {
                    log.error("\"index={}, type={}, id={}\"的文档版本冲突！", failure.getIndex(), failure.getType(), failure.getId());
                }
            }
        }
        return list;
    }

    /**
     * 根据条件搜索日志内容(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @param size       返回记录数，可以为空，默认最大返回10条。该值必须小于10000，如果超过10000请使用  {@link #queryAllByConditions}
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> queryByConditions(String indexName, String typeName, String level, String messageKey, Long startTime, Long endTime, Integer size) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        if (StringUtils.isEmpty(level) && StringUtils.isEmpty(messageKey)) {
            log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return resultList;
        }

        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.timeout(TimeValue.timeValueMinutes(2L));
        searchBuilder.query(query);
        searchBuilder.sort(order);
        if (Objects.nonNull(size)) {
            searchBuilder.size(size);
        }

        SearchRequest request = new SearchRequest(indexName).types(typeName);
        request.source(searchBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            log.error("部分分片副本处理失败！");
            for (ShardSearchFailure failure : response.getShardFailures()) {
                String reason = failure.reason();
                log.error("分片处理失败原因：{}", reason);
            }
        }
        List<Map<String, Object>> list = parseSearchResponse(response);
        if (!list.isEmpty()) {
            resultList.addAll(list);
        }
        return resultList;
    }

    private QueryBuilder generateQuery(String level, String messageKey, Long startTime, Long endTime) {
        // term query(检索level)
        TermQueryBuilder levelQuery = null;
        if (!StringUtils.isEmpty(level)) {
            levelQuery = QueryBuilders.termQuery("level", level.toLowerCase());
        }
        // match query(检索message)
        MatchQueryBuilder messageQuery = null;
        if (!StringUtils.isEmpty(messageKey)) {
            messageQuery = QueryBuilders.matchQuery("name", messageKey);
        }
        // range query(检索timestamp)
        RangeQueryBuilder timeQuery = QueryBuilders.rangeQuery(TIMESTAMP);
        timeQuery.format("epoch_millis");
        if (Objects.isNull(startTime)) {
            if (Objects.isNull(endTime)) {
                timeQuery = null;
            } else {
                timeQuery.lte(endTime);
            }
        } else {
            if (Objects.isNull(endTime)) {
                timeQuery.gte(startTime);
            } else {
                timeQuery.gte(startTime).lte(endTime);
            }
        }
        // 将上述三个query组合
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        if (Objects.nonNull(levelQuery)) {
            boolQuery.must(levelQuery);
        }
        if (Objects.nonNull(messageQuery)) {
            boolQuery.must(messageQuery);
        }
        if (Objects.nonNull(timeQuery)) {
            boolQuery.must(timeQuery);
        }
        return boolQuery;
    }

    private List<Map<String, Object>> parseSearchResponse(SearchResponse response) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        SearchHit[] hits = response.getHits().getHits();
        for (SearchHit hit : hits) {
            resultList.add(hit.getSourceAsMap());
        }
        return resultList;
    }

    /**
     * 查询所有文档id
     *
     * @return
     */
    public List<String> queryAllIdByIndexName(String indexName, String typeName) throws IOException {
        List<String> resultList = new ArrayList<>();
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.size(500);

        // 初始化 scroll 上下文
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        request.source(searchBuilder).scroll(scroll);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        String scrollId = response.getScrollId();
        SearchHit[] searchHits = response.getHits().getHits();
        // 把第一次scroll的数据添加到结果List中
        for (SearchHit searchHit : searchHits) {
            resultList.add(searchHit.getId());
        }
        // 通过传递scrollId循环取出所有相关文档
        while (searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = response.getScrollId();
            searchHits = response.getHits().getHits();
            // 循环添加剩下的数据
            for (SearchHit searchHit : searchHits) {
                resultList.add(searchHit.getId());
            }
        }
        // 清理 scroll 上下文
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        return resultList;
    }

    /**
     * 根据条件，搜索全部符合的记录(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @return
     */
    public List<Map<String, Object>> queryAllByConditions(String indexName, String typeName, String level, String messageKey, Long startTime, Long endTime) throws IOException {
        List<Map<String, Object>> resultList = new ArrayList<>();
        if (StringUtils.isEmpty(level) && StringUtils.isEmpty(messageKey)) {
            log.error("参数level(日志级别)和messageKey(日志信息关键字)不能同时为空！");
            return resultList;
        }

        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.query(query).sort(order);
        searchBuilder.size(500);

        // 初始化 scroll 上下文
        SearchRequest request = new SearchRequest(indexName).types(typeName);
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        request.source(searchBuilder).scroll(scroll);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        String scrollId = response.getScrollId();
        SearchHit[] searchHits = response.getHits().getHits();
        // 把第一次scroll的数据添加到结果List中
        for (SearchHit searchHit : searchHits) {
            resultList.add(searchHit.getSourceAsMap());
        }
        // 通过传递scrollId循环取出所有相关文档
        while (searchHits != null && searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = response.getScrollId();
            searchHits = response.getHits().getHits();
            // 循环添加剩下的数据
            for (SearchHit searchHit : searchHits) {
                resultList.add(searchHit.getSourceAsMap());
            }
        }
        // 清理 scroll 上下文
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        return resultList;
    }

    /**
     * 根据条件做分页查询(参数level和messageKey不能同时为空)
     *
     * @param level      日志级别，可以为空
     * @param messageKey 日志信息关键字，可以为空
     * @param startTime  日志起始时间，可以为空
     * @param endTime    日志结束时间，可以为空
     * @param pageNum    当前页码，可以为空(默认设为1)
     * @param pageSize   页记录数，可以为空(默认设为10)
     * @return
     * @throws IOException
     */
    public SearchPageBean<Map<String, Object>> queryPageByConditions(String indexName, String level, String messageKey, Long startTime, Long endTime, Integer pageNum, Integer pageSize) throws IOException {
        if (StringUtils.isEmpty(level) && StringUtils.isEmpty(messageKey)) {
            log.error("参数level(日志级别)、messageKey(日志信息关键字)不能同时为空！");
            return null;
        }

        if (Objects.isNull(pageNum)) {
            pageNum = 1;
        }
        if (Objects.isNull(pageSize)) {
            pageSize = 10;
        }
        QueryBuilder query = generateQuery(level, messageKey, startTime, endTime);
        FieldSortBuilder order = SortBuilders.fieldSort(TIMESTAMP).order(SortOrder.DESC);
        SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
        searchBuilder.timeout(TimeValue.timeValueMinutes(2L));
        searchBuilder.query(query);
        searchBuilder.sort(order);
        searchBuilder.from(pageNum - 1).size(pageSize);

        SearchRequest request = new SearchRequest(indexName);//.types(typeName);
        request.source(searchBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        //int totalRecord = (int) hits.getTotalHits();
        int totalRecord = (int) hits.getTotalHits().value; //会精度丢失
        List<Map<String, Object>> results = new ArrayList<>();
        for (SearchHit hit : hits.getHits()) {
            results.add(hit.getSourceAsMap());
        }

        SearchPageBean<Map<String, Object>> page = new SearchPageBean<>();
        page.setPageNo(pageNum);
        page.setPageSize(pageSize);
        page.setTotalPages(totalRecord);
        page.setData(results);
        return page;
    }


    public Object search(String index, String keywords) {
        SearchRequest request = new SearchRequest(index);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

        //分词
        List<String> analyzedWords = IkAnalzyerUtil.segmentPhraseByIk(keywords);
        for (String word : analyzedWords) {
            //boolQueryBuilder.must(matchQuery("goodsInfoName", keywords));

            boolQueryBuilder.must(queryStringQuery("" + word + "").field("goodsInfoName"));

        }
        sourceBuilder.query(boolQueryBuilder);

        request.source(sourceBuilder);

        SearchHit[] hits = new SearchHit[0];
        try {
            hits = client.search(request, RequestOptions.DEFAULT).getHits().getHits();
        } catch (Exception e) {
            log.error("ES查询出错: {}", e.getMessage(), e);
        }
        if (null != hits && hits.length > 0) {
            List<EsGoodsInfo> list = new ArrayList<>();
            for (int i = 0; i < hits.length; i++) {
                //hits[i].getSourceAsString();
                EsGoodsInfo info = JSON.parseObject(hits[i].getSourceAsString(), EsGoodsInfo.class);
                list.add(info);
            }
            return list;
        }

        return hits;
    }


    /**
     * 搜索方法
     *
     * @param index
     * @param keywords
     * @param sort
     * @param pageNo
     * @param pageSize
     * @return
     */
    public PageResult<TopicEsInfo> searchTopic(String index, String keywords, String sort, int pageNo, int pageSize) {
        if (pageNo < 1) {
            pageNo = 1;
        }
        if (pageSize > 100 || pageSize < 1) {
            pageSize = 10;
        }
        PageResult<TopicEsInfo> pageResult = new PageResult<>();

        SearchRequest request = new SearchRequest(index);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

        //分词
        List<String> analyzedWords = IkAnalzyerUtil.segmentPhraseByIk(keywords);
        for (String word : analyzedWords) {
            //boolQueryBuilder.must(matchQuery("goodsInfoName", keywords));

            boolQueryBuilder.must(queryStringQuery(word).field("title").field("navText").field("navText"));

            //boolQueryBuilder.must(QueryBuilders.multiMatchQuery(word,"title")); //不能部分匹配

        }
        sourceBuilder.query(boolQueryBuilder);
        //分页
        sourceBuilder.from((pageNo - 1) * pageSize).size(pageSize);

        if (StringHelper.isNullOrEmptyString(sort)) {
            sourceBuilder.sort("_id", SortOrder.ASC);
        } else {
            switch (sort) {
                default:
                case "asc":
                    sourceBuilder.sort("_id", SortOrder.ASC);
                    break;
                case "desc":
                    sourceBuilder.sort("_id", SortOrder.DESC);
                    break;
            }
        }
        if (!StringHelper.isNullOrEmptyString(sort)) {
            //sourceBuilder.sort("_id", SortOrder.ASC);
        }

        log.debug("es查询语句：{}", sourceBuilder.toString());

        request.source(sourceBuilder);

        SearchHit[] hits = new SearchHit[0];
        try {
            //hits = client.search(request, RequestOptions.DEFAULT).getHits().getHits();  //只取数据，没有分页信息
            SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
            if (null != searchResponse && null != searchResponse.getHits() &&
                    null != searchResponse.getHits().getHits() && searchResponse.getHits().getHits().length > 0) {
                pageResult.setRecordCount((int) searchResponse.getHits().getTotalHits().value);
                hits = searchResponse.getHits().getHits();
            }

        } catch (Exception e) {
            log.error("ES查询出错: {}", e.getMessage(), e);
        }
        if (null != hits && hits.length > 0) {
            List<TopicEsInfo> list = new ArrayList<>();
            for (int i = 0; i < hits.length; i++) {
                //hits[i].getSourceAsString();
                TopicEsInfo info = JSON.parseObject(hits[i].getSourceAsString(), TopicEsInfo.class);
                list.add(info);
            }
            pageResult.setResult(list);
            return pageResult;
        }

        return pageResult;
    }


    /**
     * 通用查询
     * 复杂查询可以自定义实现RestHighLevelClient
     *
     * @param indexName           索引
     * @param eSMappingType       返回的对象类型
     * @param searchSourceBuilder 查询条件
     * @param <T>                 泛型
     * @return pageResult
     */
    public <T> PageResult<T> search(String indexName, Class<T> eSMappingType, SearchSourceBuilder searchSourceBuilder) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new RuntimeException("indexName 不能为空");
        }

        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);

        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            PageResult<T> pageResult = new PageResult<>();
            pageResult.setRecordCount((int) hits.getTotalHits().value);
            List<T> list = new ArrayList<>();
            for (SearchHit hit : hitsArray) {
                //list.add(JsonHelper.parseBean(hit.getSourceAsString(), eSMappingType));
                list.add(JSON.parseObject(hit.getSourceAsString(), eSMappingType));
            }
            pageResult.setResult(list);
//            log.info("es查询语句：{}" , searchSourceBuilder.toString());
            return pageResult;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return new PageResult<>();
    }


    public <T> PageResult<T> searchMore(String indexName, Class<T> eSMappingType, SearchSourceBuilder searchSourceBuilder) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new RuntimeException("indexName 不能为空");
        }

        searchSourceBuilder.from(0).size(10000);

        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);

        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            PageResult<T> pageResult = new PageResult<>();
            pageResult.setRecordCount((int) hits.getTotalHits().value);
            List<T> list = new ArrayList<>();
            for (SearchHit hit : hitsArray) {
                list.add(JSON.parseObject(hit.getSourceAsString(), eSMappingType));
            }
            pageResult.setResult(list);
//            log.info("es查询语句：{}" , searchSourceBuilder.toString());
            return pageResult;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return new PageResult<>();
    }


    private XContentBuilder generateBuilder() throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        builder.startObject("properties");
        builder.startObject("message");
        builder.field("type", "text");
        // 为message字段，设置分词器为 ik_smart(最粗粒度)
        builder.field("analyzer", "ik_smart");
        builder.endObject();
        builder.startObject(TIMESTAMP);
        builder.field("type", "date");
        // 设置 日志时间的格式为  毫秒数的long类型
        builder.field("format", "epoch_millis");
        builder.endObject();
        builder.endObject();
        builder.endObject();
        return builder;
    }


    /**
     * 根据IESMappingType 的实现类,生成创建mappingType的 XContentBuilder
     *
     * @param esMapTypeClass
     * @return
     * @throws IOException
     */
    public XContentBuilder buildXContentBuilder(String index, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        XContentBuilder builder = XContentFactory.jsonBuilder().startObject();

        // 处理对象属性类型
        dealObj(builder, esMapTypeClass);
        builder.endObject();
        return builder;
    }

    /**
     * 根据对象的es属性注解判断是否进行属性处理
     *
     * @param builder
     * @param esMapTypeClass
     * @throws IOException
     */
    private void dealObj(XContentBuilder builder, Class<? extends IESMappingType> esMapTypeClass) throws IOException {
        // 子属性放到该索引下
        builder.startObject("properties");
        for (Field field : esMapTypeClass.getDeclaredFields()) {
            ESField fieldProp = field.getAnnotation(ESField.class);
            if (fieldProp != null) {
                dealField(builder, field, fieldProp);
            }
        }

        //判断对象是否有 @Suggestable 注解,有则添加 suggest 字段
        EnableSuggest suggest = esMapTypeClass.getAnnotation(EnableSuggest.class);
        if (suggest != null) {
            String analyzer = suggest.analyzerType().name();
            String suggestName = suggest.suggestName();
            builder.startObject(suggestName)
                    .field("type", "completion")
                    .field("index_analyzer", analyzer)
                    .field("search_analyzer", analyzer)
                    .field("payloads", "true")
                    .field("preserve_position_increments", false)
                    .field("preserve_separators", false)
                    .endObject();
        }
        builder.endObject();
    }


    /**
     * 处理对象的属性类型
     *
     * @param builder
     * @param field     属性
     * @param fieldProp 属性注解信息
     * @throws IOException
     */
    private void dealField(XContentBuilder builder, Field field, ESField fieldProp) throws IOException {
        try {
            if (List.class.isAssignableFrom(field.getType()) || field.getType().isArray()) {
                //list 类型的  嵌套类型和对象数组
                builder.startObject(field.getName());// 这里如果是startArray就会有问题.
                if (fieldProp.fieldType() == ESFieldType.NESTED) {
                    //嵌套类型(要查询对象信息)
                    builder.field("type", ESFieldType.NESTED.getTypeValue());
                } else {
                    //对象数组(只是保存对象信息)
                    builder.field("type", ESFieldType.OBJECT.getTypeValue());
                }

                String className = "";
                Type fc = field.getGenericType();
                if (fc instanceof ParameterizedType) {
                    ParameterizedType pt = (ParameterizedType) fc;
                    className = pt.getActualTypeArguments()[0].toString().replace("class ", "");
                } else if (field.getType().isArray()) {
                    className = field.getGenericType().toString().replace("class [L", "")
                            .replace("/", ".").replace(";", "");
                }

                Class clazz = Class.forName(className);

                if (IESMappingType.class.isAssignableFrom(clazz) || clazz.getAnnotation(ESDocObject.class) != null) {
                    dealObj(builder, clazz);
                } else if (clazz.isPrimitive() || isSimpleType(clazz)) {
                    builder.field("type", ESFieldType.STRING.getTypeValue())
                            .field("index", ESAnalyzer.not_analyzed.name()).field("store", fieldProp.isStore());
                }
                builder.endObject();
            } else if (Map.class.isAssignableFrom(field.getType())) {
                System.out.println("Map:" + field.getName());
            } else {
                // 处理简单对象
                if (isSimpleType(field.getType())) {
                    dealSimpleObjField(builder, field.getName(), fieldProp);
                    return;
                }

                // 如果是复杂的组合类型，继承于ESMapTypeI,则进行递归处理
                String className = field.getGenericType().toString().replace("class ", "");
                Class complexClazz = Class.forName(className);
                if (IESMappingType.class.isAssignableFrom(complexClazz)) {
                    builder.startObject(field.getName());
                    if (fieldProp.fieldType() == ESFieldType.NESTED) {
                        builder.field("type", ESFieldType.NESTED.getTypeValue());
                    }
                    dealObj(builder, complexClazz);
                    builder.endObject();
                }
            }
        } catch (Exception e) {
            log.error("创建mapping出错...", e);
        }
    }

    /**
     * 判断是否是简单的对象.
     *
     * @param cls
     * @return
     */
    private static boolean isSimpleType(Class cls) {
        if (cls == String.class || cls == Integer.class || cls == BigDecimal.class || cls == Date.class || cls == int.class || cls == long.class || cls == Long.class) {
            return true;
        } else {
            return false;
        }
    }

    /**
     * 处理对象类型的域值,处理已经是最简单对象的field
     */
    public void dealSimpleObjField(XContentBuilder mapbuilder, String fieldName, ESField eSMapType) throws IOException {

        mapbuilder.startObject(fieldName).field("store", eSMapType.isStore()).field("type", eSMapType.fieldType().getTypeValue());
        ESAnalyzer esAnalyzer = eSMapType.analyzerType();
        if (esAnalyzer != ESAnalyzer.not_analyzed) {
            if (esAnalyzer == ESAnalyzer.analyzed) {
                mapbuilder.field("index", "true");
            } else {
                mapbuilder.field("index", "true").
                        field("search_analyzer", esAnalyzer.name())
                        .field("analyzer", esAnalyzer.name());
            }

        } else if (esAnalyzer == ESAnalyzer.not_analyzed) {
            mapbuilder.field("index", "false");
        }
        mapbuilder.endObject();

    }

    /**
     * es聚合分组求和
     *
     * @param indexName 索引名称
     * @param groupKey  分组的key
     * @param sumKey    求和的key
     * @return Map
     */
    public Map<String, Long> groupKeySum(String indexName, SearchSourceBuilder searchSourceBuilder, String groupKey, String sumKey) {
        try {
            SearchRequest searchRequest = new SearchRequest()
                    .indices(indexName)
                    //.types(eSMappingTypeName)
                    .source(searchSourceBuilder);

            String aggName = groupKey + "_key";//分组字段别名
            String aggField = sumKey + "_key";//求和字段别名


            //分组groupKey,求和sumKey
            TermsAggregationBuilder field = AggregationBuilders.terms(aggName).field(groupKey);
            field.subAggregation(AggregationBuilders.sum(aggField).field(sumKey));
            searchSourceBuilder.aggregation(field);

            searchRequest.source(searchSourceBuilder);
            //执行查询
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);

            //获取搜索的文档结果
            Map<String, Aggregation> aggMap = response.getAggregations().getAsMap();
            Terms gradeTerms = (Terms) aggMap.get(aggName);

            Map<String, Long> returnMap = new HashMap<>();
            Map<String, Long> valueResult = new LinkedHashMap<>();
            if (null != gradeTerms.getBuckets() && gradeTerms.getBuckets().size() > 0) {
                for (Terms.Bucket bucket : gradeTerms.getBuckets()) {
                    double num = ((Sum) bucket.getAggregations().get(aggField)).getValue();
                    returnMap.put(bucket.getKeyAsString(), new Double(num).longValue());
                }

                //排序
                returnMap.entrySet().stream()
                        .sorted(Map.Entry
                                .comparingByValue(Comparator.reverseOrder()))
                        .forEachOrdered(b -> valueResult.put(b.getKey(), b.getValue()));
                return valueResult;
            }
            return new HashMap<>();

        } catch (Exception ex) {
            ex.printStackTrace();
            return null;
        }
    }


    /**
     * 查询，包含下钻聚合数据
     *
     * @param indexName           索引
     * @param eSMappingType       返回的对象类型
     * @param searchSourceBuilder 查询条件
     * @return pageResult
     */
    public IndexSearchResponse searchGroupBy(String indexName, Class eSMappingType, SearchSourceBuilder searchSourceBuilder, MallEsSearchFormDTO form) {
        if (eSMappingType == null) {
            throw new IllegalArgumentException("eSMappingType can not be null");
        }
        if (indexName == null) {
            throw new RuntimeException("indexName 不能为空");
        }

        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .types(eSMappingType.getSimpleName())
                .source(searchSourceBuilder);

        try {
            SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = search.getHits();
            SearchHit[] hitsArray = hits.getHits();
            // 封装返回结果
            IndexSearchResponse response = new IndexSearchResponse();
            // 放入源数据
            response.setOriginHits(hits);

            List<String> sources = new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
                sources.add(hit.getSourceAsString());
            }
            response.putReslutData(sources);
            // 聚合结果
            response.putResultAggs(agg2Map(search.getAggregations()));
            // 总数据
            response.setPageNum(form.getPageNo());
            response.setPageSize(form.getPageSize());
            response.setTotalCount((int) hits.getTotalHits().value);
            return response;
        } catch (Exception e) {
            log.error("查询文档:{},失败:{}", indexName, e.getMessage());
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 处理一个聚合对象数据{@link Aggregations}
     *
     * @param aggregations
     * @return Map{key:一个聚合的名称,value:聚合后的桶数据}
     */
    public Map<String, Object> agg2Map(Aggregations aggregations) {
        Map<String, Object> resultMap = new HashMap<>();
        if (aggregations == null) {
            return resultMap;
        }
        for (Aggregation aggregation : aggregations) {

            resultMap.put(aggregation.getName(), dealOneAggregation(aggregation));
        }
        return resultMap;
    }

    /**
     * 处理一个聚合对象
     *
     * @param aggregation {@link Aggregation}
     * @return 如果只有一个桶数据, 则返回对象, 如果有多个桶数据, 则返回数组
     */
    public Object dealOneAggregation(Aggregation aggregation) {
        if (aggregation instanceof ParsedStringTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedStringTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedDoubleTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedDoubleTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedLongTerms) {
            Collection<? extends Terms.Bucket> buckets = ((ParsedLongTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof UnmappedTerms) {
            Collection<? extends Terms.Bucket> buckets = ((UnmappedTerms) aggregation).getBuckets();
            return dealBunkets(buckets);
        } else if (aggregation instanceof ParsedNested) {
            Aggregations aggregations = ((ParsedNested) aggregation).getAggregations();
            return agg2list(aggregations);
        } else if (aggregation instanceof ParsedAvg) {
            return aggregation.getName();
        } else {
            throw new IllegalArgumentException("未知聚合类型,不可处理");
        }
    }

    /**
     * 处理一个聚合下边的一个或多个桶数据
     *
     * @param buckets
     * @return 如果有一个桶, 那么就返回一个对象, 而不是列表 如果有多个桶,则返回列表 如果没有值,则返回一个空对象
     */
    public Object dealBunkets(Collection<? extends Terms.Bucket> buckets) {
        List<Object> list = new ArrayList<>();
        for (Terms.Bucket bucket : buckets) {
            list.add(dealOneBunket(bucket));
        }
        return list.isEmpty() ? new Object() : list.size() == 1 ? list.get(0) : list;
    }

    /**
     * 处理一个bunkey
     *
     * @param bucket
     * @return 如果没有子查询返回bunkey中的值[String]
     * 如果有子查询,返回一个对应的map对象[HashMap]--{key:bunketKey,value:子查询返回的map}
     */
    public Object dealOneBunket(Terms.Bucket bucket) {
        Object params = null;
        if (bucket.getAggregations().iterator().hasNext()) {
            params = agg2list(bucket.getAggregations());
        }
        if (params == null) {
            return bucket.getKey();// 没有子查询
        } else if (params instanceof List) {
            List<Object> resultList = (List) params;
            if (resultList.size() == 1) {
                return resultList.get(0);
            } else if (resultList.size() > 1) {
                Map<String, Object> resultMap = new HashMap<>();
                resultMap.put(bucket.getKey().toString(), resultList);
                return resultMap;
            } else {
                return resultList;
            }
        } else {
            // 没有子查询
            Map<String, Object> resultMap = new HashMap<>();
            resultMap.put(bucket.getKey().toString(), params);
            return resultMap;
        }
    }

    /**
     * 处理子聚合的方法
     *
     * @param aggregations
     * @return
     * @author of1081
     */
    public Object agg2list(Aggregations aggregations) {
        List<Object> resultList = new ArrayList<>();
        for (Aggregation aggregation : aggregations) {
            if (!(aggregation instanceof InternalAvg)) {
                resultList.add(dealOneAggregation(aggregation));
            }
        }
        return resultList.isEmpty() ? new Object() : resultList.size() == 1 ? resultList.get(0) : resultList;
    }


}

